package consumer_test

import (
	"context"
	"fmt"
	"gitee.com/zaper/kafka-service/model"
	"gitee.com/zaper/kafka-service/service/connect"
	metadata "gitee.com/zaper/kafka-service/service/metadata"
	"gitee.com/zaper/kafka-service/util"
	"gitee.com/zaper/private-logger"
	"github.com/petermattis/goid"
	"github.com/segmentio/kafka-go"
	"math/rand"
	"sync"
	"time"
)

var (
	sw sync.RWMutex
)

// ExecClientConsumerMess 使用client.Fetch的消费者模型
func ExecClientConsumerMess() {
	conn := connect.GetKafkaConnectManager()
	ctx := context.Background()
	// 主题和分区信息
	topicName := "TestTopic"
	var partitions = metadata.GetMetadataManager().GetAllPartitionsByTopicName(topicName)
	fmt.Println("开始消费消息...")
	// 构建 Fetch 请求
	// 发送 Fetch 请求
	for {
		req := &kafka.FetchRequest{
			Topic:     topicName,
			Partition: rand.Intn(len(partitions)),
		}
		resp, err := conn.Client.Fetch(ctx, req)
		if err != nil {
			logger.Infof("Fetch 失败: %v", err)
			time.Sleep(1 * time.Second) // 重试前等待
		}
		rd, _ := resp.Records.ReadRecord()
		rt := make([]byte, 1024)
		_, _ = rd.Value.Read(rt)
		logger.Info(resp.LastStableOffset, string(rt))
	}
}

// execReaderMess 使用Reader的消费者模型
func execReaderMess(kafkaModel *model.KafkaConsumerModel) {
	logger.Infof("%d号消费者进入%s消费组", goid.Get(), kafkaModel.GroupId)
	// 创建Kafka读取器(最简配置，优化项配置参考tmp目录consumer的模型)
	brokerAddr := []string{"localhost:9091"}
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:         brokerAddr,
		GroupID:         kafkaModel.GroupId, // 指定消费者组ID
		GroupTopics:     kafkaModel.Topics,
		StartOffset:     kafka.FirstOffset,
		ReadLagInterval: 5 * time.Second,
	})
	defer util.CloseWRIo(&util.KafkaWRIo{Reader: r})
	ctx, cancel := context.WithCancel(context.Background())
	go monitorExitSign(cancel, kafkaModel.StopChan[goid.Get()])
	for {
		select {
		case <-ctx.Done():
			logger.Info("cancel be exec customer shutting down")
			// 已经确认要下线的了就移除它
			deleteSignMap(kafkaModel.StopChan, goid.Get())
			return
		default:
		}
		msg, err := r.ReadMessage(ctx)
		if err != nil {
			logger.Infof("消费错误: %v（将重试）", err)
			time.Sleep(200 * time.Millisecond)
			continue
		}
		err = r.CommitMessages(ctx, msg) // 手动提交，防止不期望的现象
		// 打印消息详情
		logger.Infof("消费者：主题=%s | 分区=%d | 偏移量=%d | 消费组=%s | %d号消费者读取到：%s", msg.Topic,
			msg.Partition, msg.Offset, kafkaModel.GroupId, goid.Get(), string(msg.Value))
	}
}

func monitorExitSign(cancel context.CancelFunc, ch chan struct{}) {
	defer close(ch)
	select {
	case <-ch:
		logger.Info("kafka segment customer shutting down")
		cancel()
		return
	}
}
